package cn.neo.hhrpc.core.consumer;

import cn.neo.hhrpc.core.api.*;
import cn.neo.hhrpc.core.consumer.http.OkHttpInvoker;
import cn.neo.hhrpc.core.exception.RpcException;
import cn.neo.hhrpc.core.governance.SlidingTimeWindow;
import cn.neo.hhrpc.core.meta.InstanceMeta;
import lombok.extern.slf4j.Slf4j;

import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.net.SocketTimeoutException;
import java.util.*;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

import static cn.neo.hhrpc.core.util.MethodUtils.checkLocalMethod;
import static cn.neo.hhrpc.core.util.MethodUtils.methodSign;
import static cn.neo.hhrpc.core.util.TypeUtils.castMethodResult;

/**
 * 消费端动态代理处理类
 */
@Slf4j
public class HHInvocationHandler implements InvocationHandler {


    Class<?> service;

    RpcContext context;

    List<InstanceMeta> providers;

    // 隔离列表
    final Set<InstanceMeta> isolateProviders = new LinkedHashSet<>();

    // 探活列表
    final List<InstanceMeta> halfOpenProviders = new ArrayList<>();

    ScheduledExecutorService executor;

    // 通过滑动窗口统计(X时间内出现N次异常才算失败)出现故障的实例
    Map<String, SlidingTimeWindow> windows = new HashMap<>();

    HttpInvoker httpInvoker;

    public HHInvocationHandler(Class<?> service, RpcContext context, List<InstanceMeta> providers) {
        this.service = service;
        this.context = context;
        this.providers = providers;
        int timeout = Integer.parseInt(context.getParameters().getOrDefault("consumer.timeout", "1000"));
        this.httpInvoker = new OkHttpInvoker(timeout);
        int halfOpenInitialDelay = Integer.parseInt(context.getParameters().getOrDefault("consumer.halfOpenInitialDelay", "10000"));
        int halfOpenDelay = Integer.parseInt(context.getParameters().getOrDefault("consumer.halfOpenDelay", "60000"));
        this.executor = Executors.newScheduledThreadPool(1);
        // 延迟10s, 每60s执行一次，
        this.executor.scheduleWithFixedDelay(this::halfOpen, halfOpenInitialDelay, halfOpenDelay, TimeUnit.MILLISECONDS);
//        this.executor.scheduleWithFixedDelay(this::halfOpen, 10, 60, TimeUnit.SECONDS);
    }

    private void halfOpen() {
        log.info(" ===> half open isolatedProviders: " + isolateProviders);
        halfOpenProviders.clear();
        halfOpenProviders.addAll(isolateProviders);
    }

    /**
     * 调用provider
     */
    @Override
    public Object invoke(Object proxy, Method method, Object[] args) {

        if (checkLocalMethod(method)) {
            return null;
        }

        RpcRequest rpcRequest = new RpcRequest();
        rpcRequest.setService(service.getCanonicalName());
        rpcRequest.setMethodSign(methodSign(method));
        rpcRequest.setArgs(args);

        // 重试次数
        int retries = Integer.parseInt(context.getParameters().getOrDefault("consumer.retries", "1"));
        // 故障次数
        int faultLimit = Integer.parseInt(context.getParameters().getOrDefault("consumer.faultLimit", "10"));

        while (retries-- > 0) {

            log.info(" ===> 当前重试次数retries: " + retries);
            try {
                // CacheFilter
                // 前置过滤器，期望返回上次调用的响应值
                // 如果当前请求，跟上次请求一样，则返回上次的调用结果，不进行后续的运行
                for (Filter filter : this.context.getFilters()) {
                    Object preResult = filter.prefilter(rpcRequest);
                    if (null != preResult) {
                        log.debug(filter.getClass().getName() + " ===> prefilter:{}", preResult);
                        return preResult;
                    }
                }

                InstanceMeta instance;
                synchronized (halfOpenProviders) {
                    if (halfOpenProviders.isEmpty()) {
                        List<InstanceMeta> instances = context.getRouter().route(providers);
                        instance = context.getLoadBalancer().choose(instances);
                        log.info("loadBalancer.choose(nodes) ====> {}", instance);
                    } else {
                        instance = halfOpenProviders.remove(0);
                        log.info(" check alive instance ===> {}", instance);
                    }
                }

                RpcResponse<?> rpcResponse;
                Object result;
                String url = instance.toUrl();

                // TODO BUG:这里拿到的可能不是最终值，可放在请求结束后拿到最终值再处理
//        for (Filter filter : this.context.getFilters()) {
//            Object postResult = filter.postfilter(rpcRequest, rpcResponse, result);
//            if (null != postResult) {
//                return postResult;
//            }
//        }

                /**
                 * 此处做故障隔离
                 */
                try {
                    rpcResponse = httpInvoker.post(rpcRequest, instance.toUrl());
                    result = castReturnResult(method, rpcResponse);
                } catch (Exception e) {
                    // 故障的规则统计和隔离
                    // 每一次异常，记录一次，统计30s的异常
                    SlidingTimeWindow window = windows.get(url);
                    if (window == null) {
                        window = new SlidingTimeWindow();
                        windows.put(url, window);
                    }

                    // 记录当前时间点发生了1次异常
                    window.record(System.currentTimeMillis());
                    log.info("instance {} in window with {}", url, window.getSum());

                    // 发生10次，就做故障隔离
                    if (window.getSum() >= faultLimit) {
                        isolate(instance);
                    }
                    throw e;
                }

                // 判断当前请求是否是探活
                synchronized (providers) {
                    if (!providers.contains(instance)) {
                        isolateProviders.remove(instance);
                        providers.add(instance);
                        log.info(" ===> instance {} is recovered. isolateProviders={},providers={}", instance, isolateProviders, providers);
                    }
                }

                // 后置过滤器：将当前请求的结果放到过滤器中，用于前置过滤器的判断
                for (Filter filter : this.context.getFilters()) {
                    Object postResult = filter.postfilter(rpcRequest, rpcResponse, result);
                    if (null != postResult) {
                        return postResult;
                    }
                }
                return result;
            } catch (Exception ex) {
                // 超时异常则重试
                if (!(ex.getCause() instanceof SocketTimeoutException)) {
                    throw ex;
                }
            }
        }

        return null;

    }

    /**
     * 故障隔离
     * @param instance
     */
    private void isolate(InstanceMeta instance) {
        log.info(" ===> isolate instance: " + instance);
        providers.remove(instance);
        log.info(" ===> providers = {}" + providers);
        isolateProviders.add(instance);
        log.info(" ===> isolateProviders = {}" + isolateProviders);

    }

    /**
     * 将RPC响应结果转换为对应的方法返回值。
     * @param method 执行的RPC方法，用于指导结果的转换。
     * @param rpcResponse RPC调用的响应对象，包含调用结果或异常信息。
     * @return 方法的返回值，类型依据实际调用结果动态转换。
     */
    private static Object castReturnResult(Method method, RpcResponse<?> rpcResponse) {
        if (rpcResponse.isStatus()) {
            return castMethodResult(method, rpcResponse.getData());
        } else {
            RpcException exception = rpcResponse.getEx();
            if (exception != null) {
                log.error("response error.", exception);
                throw exception;
            }
            return null;
        }
    }


}
